package com.education.realtime.app;

import com.education.realtime.common.Constant;
import com.education.realtime.util.FlinkSourceUtil;
import com.education.realtime.util.SqlUtil;
import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.runtime.state.hashmap.HashMapStateBackend;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.api.common.time.Time;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.HashMap;
import java.util.Map;

public abstract class BaseAppV2 {
    public void init(int port, String ckGroupIdJobName, String firstTopic, String ... otherTopics){

        System.setProperty("HADOOP_USER_NAME", "atguigu");
        Configuration conf = new Configuration();
        conf.setInteger("rest.port",port);
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);
        env.setRestartStrategy(RestartStrategies.failureRateRestart(10, Time.days(1),Time.seconds(5)));
//        env.setStateBackend(new HashMapStateBackend());
//        env.getCheckpointConfig().setCheckpointStorage("hdfs://hadoop102:8020/gmall/" + ckGroupIdJobName);
//        env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
//        // 每 3000ms 开始一次 checkpoint
//        env.enableCheckpointing(3000);
//        // Checkpoint 必须在一分钟内完成，否则就会被抛弃
//        env.getCheckpointConfig().setCheckpointTimeout(60 * 1000);
//        // 同一时间只允许一个 checkpoint 进行
//        env.getCheckpointConfig().setMaxConcurrentCheckpoints(1);
//        // 确认 checkpoints 之间的时间会进行 500 ms
//        env.getCheckpointConfig().setMinPauseBetweenCheckpoints(500);
//        // 使用 externalized checkpoints，这样 checkpoint 在作业取消后仍就会被保留
//        env.getCheckpointConfig().setExternalizedCheckpointCleanup(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);

        ArrayList<String> topics = new ArrayList<>();
        if (firstTopic != null) {
            topics.add(firstTopic);
            topics.addAll(Arrays.asList(otherTopics));
        }
        Map<String, DataStreamSource<String>> topicAndStream = new HashMap<>();
        for (String topic : topics) {
            DataStreamSource<String> stream = env.addSource(FlinkSourceUtil.getKafkaSource(ckGroupIdJobName, topic));
            topicAndStream.put(topic,stream);
        }

        handle(env,topicAndStream);

        try {
            env.execute(ckGroupIdJobName);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    protected abstract  void handle(StreamExecutionEnvironment env, Map<String, DataStreamSource<String>> stream) ;


}

